package com.amigo.online.provider.search.rocketmq.config;


import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.alibaba.fastjson.JSON;
import com.amigo.online.provider.search.rocketmq.PushMessage;
import com.amigo.online.provider.search.rocketmq.model.BaseMqSendModel;

/**
 * @Author:kml
 * @Description:
 */
@Configuration
@EnableConfigurationProperties(RocketMqProperties.class)
public class RocketMqConfiguration implements ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(RocketMqConfiguration.class);
    private static ApplicationContext context;
    @Autowired
    RocketMqProperties rocketMqProperties;

    /**
     * 初始化向rocketmq发送普通消息的生产者
     */
    @Bean(name = RocketMqProperties.DEFAULT)
    public DefaultMQProducer defaultMQProducer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(rocketMqProperties.getDefault_producer());
        producer.setNamesrvAddr(rocketMqProperties.getNamesrv_addr());
        producer.setVipChannelEnabled(false);
        producer.start();
        log.info("RocketMq defaultProducer Started.");
        return producer;
    }

    /**
     * 初始化向rocketmq发送事务消息的生产者
     */
    @Bean(name = RocketMqProperties.TRANS)
    public TransactionMQProducer transactionMQProducer() throws MQClientException {
        TransactionMQProducer producer = new TransactionMQProducer(rocketMqProperties.getTransaction_producer());
        producer.setNamesrvAddr(rocketMqProperties.getNamesrv_addr());
        producer.setTransactionCheckListener((MessageExt msg) -> {
            // 事务回查机制
            return LocalTransactionState.COMMIT_MESSAGE;
        });
        // 事务回查最小并发数
        producer.setCheckThreadPoolMinSize(2);
        // 事务回查最大并发数
        producer.setCheckThreadPoolMaxSize(5);
        // 队列数
        producer.setCheckRequestHoldMax(2000);
        producer.start();
        log.info("RocketMq TransactionMQProducer Started.");
        return producer;
    }

    /**
     * 普通消费者
     */
    @Bean
    public DefaultMQPushConsumer pushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMqProperties.getDefault_consumer());
        consumer.subscribe(rocketMqProperties.getDefault_topic(), rocketMqProperties.getTag());
        consumer.setNamesrvAddr(rocketMqProperties.getNamesrv_addr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(1);
//		consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
            MessageExt msg = msgs.get(0);
            try {
                if (msgs.size() == 0)
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                handleMessage(msg);
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            // 如果没有return success，consumer会重复消费此信息，直到success。
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        new Thread(() -> {
            try {
                Thread.sleep(10000);// 延迟10秒再启动，主要是等待spring事件监听相关程序初始化完成，否则，回出现对RocketMQ的消息进行消费后立即发布消息到达的事件，然而此事件的监听程序还未初始化，从而造成消息的丢失
                try {
                    consumer.start();
                } catch (MQClientException e) {
                    log.info("RocketMq pushConsumer Start failure!!!.");
                    e.printStackTrace();
                }
                log.info("RocketMq 普通消费者 Started.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        return consumer;
    }

    /**
     * 顺序消费者
     */
    @Bean
    public DefaultMQPushConsumer pushOrderConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMqProperties.getOrder_consumer());
        consumer.subscribe(rocketMqProperties.getOrder_topic(), rocketMqProperties.getTag());
        consumer.setNamesrvAddr(rocketMqProperties.getNamesrv_addr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeOrderlyContext context) -> {
            MessageExt msg = msgs.get(0);
            try {
                if (msgs.size() == 0) return ConsumeOrderlyStatus.SUCCESS;
                context.setAutoCommit(true);
                handleMessage(msg);
            } catch (NoSuchBeanDefinitionException e) {
                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
            // 如果没有return success，consumer会重复消费此信息，直到success。
            return ConsumeOrderlyStatus.SUCCESS;
        });
        new Thread(() -> {
            try {
                Thread.sleep(10000);
                try {
                    consumer.start();
                } catch (MQClientException e) {
                    log.info("RocketMq 顺序消费者 Started failure!!!.");
                    e.printStackTrace();
                }
                log.info("RocketMq 顺序消费者 Started.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        return consumer;
    }

    private void handleMessage(MessageExt msg) throws Exception {
//		String bodyStr = new String(msg.getBody(),"utf-8");
//		BaseMqSendModel bmsm = JSON.parseObject(bodyStr, BaseMqSendModel.class);
//		bmsm.setKeys(msg.getKeys());
//		PushMessage pushMessage = (PushMessage)context.getBean(bmsm.getBeanName());
//		pushMessage.handleMessage(bmsm);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        context = applicationContext;
    }

    ;
}
